package com.example.iot.queue.util;

import com.example.iot.entity.id.EntityId;
import com.example.iot.queue.TopicPartitionInfo;
import com.example.iot.queue.configuration.RuleEngineQueueConfig;
import com.google.common.hash.HashFunction;
import com.google.common.hash.Hashing;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

/**
 * @author ChenXiangpeng
 */
@Service
public class TopicPartitionService {
    @Value("${queue.core.partitions}")
    private Integer corePartitions;
    private final RuleEngineQueueConfig ruleEngineQueueConfig;
    /**
     * 队列对应主题名
     */
    private final ConcurrentMap<String, String> partitionTopics = new ConcurrentHashMap<>();
    /**
     * 队列对应分区数量
     */
    private final ConcurrentMap<String, Integer> partitionSizes = new ConcurrentHashMap<>();

    /**
     * 分区Hash计算规则
     */
    private static HashFunction hashFunction = Hashing.murmur3_128();
    /*public static HashFunction forName(String name) {
        switch (name) {
            case "murmur3_32":
                return Hashing.murmur3_32();
            case "murmur3_128":
                return Hashing.murmur3_128();
            case "sha256":
                return Hashing.sha256();
            default:
                throw new IllegalArgumentException("Can't find hash function with name " + name);
        }
    }*/

    public TopicPartitionService(RuleEngineQueueConfig ruleEngineQueueConfig){
        this.ruleEngineQueueConfig = ruleEngineQueueConfig;
    }

    @PostConstruct
    public void init() {
        partitionTopics.put(ruleEngineQueueConfig.getName(), ruleEngineQueueConfig.getTopic());
        partitionSizes.put(ruleEngineQueueConfig.getName(), ruleEngineQueueConfig.getPartitions());
    }

    public TopicPartitionInfo resolve(String serviceQueue, EntityId entityId) {
        int hash = hashFunction.newHasher()
                .putLong(entityId.getId().getMostSignificantBits())
                .putLong(entityId.getId().getLeastSignificantBits()).hash().asInt();

        Integer partitionSize = partitionSizes.getOrDefault(serviceQueue, corePartitions);
        int partition;
        if (partitionSize != null) {
            partition = Math.abs(hash % partitionSize);
        } else {
            //TODO: In 2.6/3.1 this should not happen because all Rule Engine Queues will be in the DB and we always know their partition sizes.
            partition = 0;
        }

        return buildTopicPartitionInfo(serviceQueue, partition);
    }

    public static TopicPartitionInfo buildTopicPartitionInfo(String serviceQueue, int partition) {
        TopicPartitionInfo.TopicPartitionInfoBuilder tpi = TopicPartitionInfo.builder();
        tpi.topic(serviceQueue);
        tpi.partition(partition);
        /*ServiceQueueKey myPartitionsSearchKey;
        if (isIsolated(serviceQueue, tenantId)) {
            tpi.tenantId(tenantId);
            myPartitionsSearchKey = new ServiceQueueKey(serviceQueue, tenantId);
        } else {
            myPartitionsSearchKey = new ServiceQueueKey(serviceQueue, new TenantId(TenantId.NULL_UUID));
        }
        List<Integer> partitions = myPartitions.get(myPartitionsSearchKey);
        if (partitions != null) {
            tpi.myPartition(partitions.contains(partition));
        } else {
            tpi.myPartition(false);
        }*/
        return tpi.build();
    }
}
